package code

import (
	"context"
	"fmt"
	pubsub2 "github.com/moby/moby/pkg/pubsub"
	"goStudy/grpc/pubsub/proto"
	"strings"
	"time"
)

type PubsubServiceServer struct{
	pub *pubsub2.Publisher
	auth *Authentication
}


func NewPubsubService() *PubsubServiceServer {
	return &PubsubServiceServer{
		pub: pubsub2.NewPublisher(100*time.Millisecond, 10),
		auth:&Authentication{
			User:     "gopher",
			Password: "password",
		},
	}
}

func (this *PubsubServiceServer) Publish(ctx context.Context, req *pubsub.String) (*pubsub.String, error) {
	if err:=this.auth.Auth(ctx);err!= nil {
		return nil,err
	}

	this.pub.Publish(req.GetValue())
	return &pubsub.String{},nil
}

func (this *PubsubServiceServer) Subscribe(req *pubsub.String, stream pubsub.PubsubService_SubscribeServer) error {
	ch:=this.pub.SubscribeTopic(func(v interface{}) bool {
		fmt.Println("发布消息:",v)
		if key, ok := v.(string); ok {
			if strings.HasPrefix(key,req.GetValue()) {
				return true
			}
		}
		return false
	})

	for v:=range ch {
		if err:=stream.Send(&pubsub.String{Value:v.(string)});err!= nil {
			return err
		}
	}

	return nil
}

